package com.song.ResourceNode.server;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.SimpleChannelInboundHandler;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import com.alibaba.fastjson.JSON;
import com.song.ResourceNode.bean.ResourceConfig;
import com.song.ResourceNode.client.NodeClient;
import com.song.ResourceNode.thread.FetchThread;
import com.song.ResourceNode.util.Utils;
import com.song.corebean.Host;
import com.song.corebean.MessageType;
import com.song.corebean.PackageHead;
import com.song.corebean.PackageMessage;
import com.song.corebean.ProsessBean;
import com.song.corebean.mapred.MapReduce;
import com.song.corebean.utils.CoreUtils;

public class NodeServerHandler extends ChannelInboundHandlerAdapter{

	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
		// TODO Auto-generated method stub
		super.channelReadComplete(ctx);
		ctx.flush();
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object obj)
			throws Exception {

		PackageMessage msg = (PackageMessage)obj;
		if(ResourceConfig.mapReduceClazz==null)
			return;
		Class clazz = Class.forName(ResourceConfig.mapReduceClazz);
		MapReduce mapred = (MapReduce) clazz.newInstance(); 
		if(ResourceConfig.isMaster && msg.head.messageType == MessageType.DEAL_ALL_DATA)
		{
			List data = (List)msg.body.bundle.get("data");
			List result = new ArrayList();
			List<Host> hosts =null;
			NodeClient client = new NodeClient();
			client.connect(ResourceConfig.infoCenterAddr,
					ResourceConfig.infoCenterPort);

			Object res = client.sendAndRecv(MessageType.GET_HOSTS,null,null);
			
			if(res!=null)
			{
				hosts = (List<Host>) res;
				List<List<ProsessBean>> datasegs =mapred.seg(data, hosts.size());
				
				ExecutorService pool = Executors.newFixedThreadPool(hosts.size());
				List<Callable<Object>> tasks = new ArrayList();
				for(int i=0; i<hosts.size();i++)
					tasks.add(new FetchThread(hosts.get(i),datasegs.get(i)));
				
				List<Future<Object>> futures  = pool.invokeAll(tasks);
				for(Future<Object> future :futures)
				{
					if(future.isDone())
					{
						result.add(future.get());
					}
				}
				pool.shutdown();
				Object reduce  =  mapred.reduce(result);
				PackageMessage sndmsg = CoreUtils.ctxToPackageMessage(ctx, MessageType.DEAL_ALL_DATA, "data",reduce);
				ctx.writeAndFlush(sndmsg);
			}
		}else{
			
			List<ProsessBean> data = (List<ProsessBean>)msg.body.bundle.get("data");
			Object res =mapred.deal(data);
			PackageMessage sndmsg = CoreUtils.ctxToPackageMessage(ctx, MessageType.DEAL_SEG_DATA, "data",res);
			System.out.println("发送数据 " +sndmsg);
			ctx.writeAndFlush(sndmsg);
		}
		

	}

	@Override
	public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
		// TODO Auto-generated method stub
		super.channelRegistered(ctx);
		System.out.println("接受连接" + ctx.channel().remoteAddress());
	}

	

}
